package t20240514_lineage;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.List;


public class Test2 {

    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tbEnv = StreamTableEnvironment.create(env);

        String sourceSql = "CREATE TABLE datagen (" +
                "    score INT," +
                "    sn STRING" +
                "  ) WITH (" +
                "    'connector' = 'datagen'," +
                "    'rows-per-second'='1'," +
                "    'fields.score.kind'='sequence'," +
                "    'fields.score.start'='1'," +
                "    'fields.score.end'='10'," +
                "    'fields.sn.length'='10'" +
                "  )";

        String sinkSql =
                "  CREATE TABLE redis_sink (" +
                        "    key STRING," +
                        "    score INT," +
                        "    sn STRING," +
                        "    mins DOUBLE," +
                        "    maxs DOUBLE" +
                        "    ) WITH (" +
                        "    'connector' = 'redis'," +
                        "    'host' = '127.0.0.1'," +
                        "    'port' = '6379'," +
                        "    'redis-mode' = 'single'," +
                        "    'timeout' = '2000'," +
                        "    'password' = 'yy2024'," +
                        "    'command' = 'ZADD'," +
                        "    'zrem.rangeby' = 'SCORE'," +
                        "    'max.retries' = '5'" +
                        "  )";

        String execSql = "INSERT INTO redis_sink select 'SN_TEST' as key,score,sn,1,5 from datagen";

        String[] sqls = new String[]{sourceSql, sinkSql, execSql};
//
//
//        LineageContext lineageContext = new LineageContext(tbEnv);
//        LineageResult lineageResult = lineageContext.analyzeLineage(sqls);
//
//        JSONObject jsonObject = (JSONObject) JSONObject.toJSON(lineageResult);
//        System.out.println(jsonObject);

        for (String sql : sqls) {
            System.out.println(sql);
        }


    }
}
